package com.tzx.fluttermixedapplication.test

import android.util.Log
import android.view.View
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.MainScope
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.reduce
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking


// Flow 是冷流，冷流不会发射数据，直到它被收集，所以冷流是“声明式的”。
// 所有能触发收集数据动作的消费者称为终端消费者，它就像点燃鞭炮的星火，使得被若干个中间消费者套娃了的流从外向内（从下游到上游）一个个的被收集，最终传导到原始流，触发数据的发射。

// 流中的数据不会自动生产，直到流被收集的那一刻。当流被收集的瞬间，数据开始生产并被发射出去，通过流收集器将其传递给消费者。
// 流和流收集器是成对出现的概念。流是一组按序产生的数据，数据的产生表现为通过流收集器发射数据，在这里流收集器像是流数据容器（虽然它不持有任何一条数据），它定义了如何将数据传递给消费者。
fun testFlow() {
    GlobalScope.launch {
        // 构建安全流（传入 block 定义如何生产流数据）
        // public fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
        // 安全流继承自抽象流
        // private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
        //    override suspend fun collectSafely(collector: FlowCollector<T>) {
        //        collector.block()// 收集流数据时调用 block，即触发生产数据
        //  }
        // 抽象流
        // public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {
        //     // 收集数据的具体实现
        //     public final override suspend fun collect(collector: FlowCollector<T>) {
        //         // 构建 FlowCollector 并传入 collectSafely()
        //         val safeCollector = SafeCollector(collector, coroutineContext)
        //         try {
        //             collectSafely(safeCollector)
        //         } finally {
        //             safeCollector.releaseIntercepted()
        //         }
        //     }
        //     public abstract suspend fun collectSafely(collector: FlowCollector<T>)
        // }
        flow {
            (1 .. 3).forEach {
                delay(1000)
                emit(it)
            }
        }.collect {
            // public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
            //     collect(object : FlowCollector<T> {
            //         override suspend fun emit(value: T) = action(value)
            //  })
            Log.d("tanzhenxing", "num=$it")
        }
    }
}

// transform() 做了三件事情：构建了一个新流（下游流），当下游流被收集时，会立马收集上游的流，当收集到上游数据后将其传递给transform这个 lambda。
// transform() 建立了一种在流上拦截并转发的机制：新建下游流，它生产数据的方式是通过收集上游数据，并将数据转发到一个带有发射数据能力的 lambda 中。
// transform() 这个中间消费者在拦截上游数据后，就可随心所欲地将其变换后再转发给下游消费者。
// public inline fun <T, R> Flow<T>.transform(
//     crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
// ): Flow<R> =
//     // 构建下游流
//     flow {
//         // 收集上游数据（这里的逻辑在下游流被收集的时候调用）
//         collect { value ->
//             // 处理上游数据
//             return@collect transform(value)
//         }
//     }
fun <T> countdown(
    duration: Long,
    interval: Long,
    onCountdown: suspend (Long) -> T
): Flow<T> =
    flow { (duration - interval downTo 0 step interval).forEach { emit(it) } }
        // onEach() 通过 transform() 构建了一个下游流，并在转发每一个上游流数据前又做了一件额外的事情，用lambda action表示。
        // public fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T> = transform { value ->
        //     action(value)
        //     return@transform emit(value)
        // }
        .onEach { delay(interval) }
        // public fun <T> Flow<T>.onStart(
        //     action: suspend FlowCollector<T>.() -> Unit
        // ): Flow<T> = unsafeFlow { // 构建下游流
        //     val safeCollector = SafeCollector<T>(this, currentCoroutineContext())
        //     try {
        //         safeCollector.action() // 在收集上游流数据之前执行动作
        //     } finally {
        //         safeCollector.releaseIntercepted()
        //     }
        //     collect(this) // 收集上游流数据
        // }
        //
        // internal inline fun <T> unsafeFlow(crossinline block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
        //     // 构建新流
        //     return object : Flow<T> {
        //         override suspend fun collect(collector: FlowCollector<T>) {
        //             collector.block()
        //         }
        //     }
        // }
        // unsafeFlow() 直接实例化了Flow接口，并定义了该流被收集时执行的操作，即调用block。
        // 所以 unsafeFlow() 和 transform 很类似，都新建下游流以收集了上游数据，只不过在收集动作（所有数据发射之前）之前做了一件额外的事。
        .onStart { emit(duration) }
        // map() 通过 transform() 构建了一个下游流，并且在拿到上游流数据时先将其进行了transform变换，然后再转发出去。
        // public inline fun <T, R> Flow<T>.map(crossinline transform: suspend (value: T) -> R): Flow<R> =
        //     transform { value -> return@transform emit(transform(value)) }
        .map { onCountdown(it) }
        .flowOn(Dispatchers.Default)

fun testReduce() {
    Log.v("tanzhenxing", "${Thread.currentThread().name} call main fun")
    val mainScope = MainScope()
    mainScope.launch {
        Log.v("tanzhenxing", "${Thread.currentThread().name} call main scope.launch")
        val ret = countdown(20_000, 1_000) { it }
            .onStart { Log.v("tanzhenxing", "${Thread.currentThread().name} countdown start") }
            .onCompletion { Log.v("tanzhenxing", "${Thread.currentThread().name} countdown end") }
            .reduce { acc, value ->
                Log.v("tanzhenxing", "${Thread.currentThread().name} call reduce fun")
                return@reduce acc + value
            } // 终端消费者：计算所有异步结果的和
        // 因为 reduce() 是一个 suspend 方法，所以会挂起协程，直到倒计时完成才打印所有异步结果的和
        Log.v("tanzhenxing", "${Thread.currentThread().name} coutdown acc ret = $ret")
    }
    Log.v("tanzhenxing", "${Thread.currentThread().name} call main fun end")
}